package com.chukun.disruptor.higher.chain;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.*;

/**
 * 主程序的运行
 */

public class TradeMainRunner {

    public static void main(String[] args) throws InterruptedException {

        int ringBufferSize = 1024*1024;
        //创建自定义的线程池
        ExecutorService executorService = Executors.newCachedThreadPool();
        Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        },ringBufferSize,executorService, ProducerType.SINGLE,new BusySpinWaitStrategy());

        /**
         *   关联消费者，串行执行
         *   只要依次添加到EventHandlerGroup里面即可

         disruptor.handleEventsWith(new EventHandlerTaskOne())
                  .handleEventsWith(new EventHandlerTaskTwo())
                  .handleEventsWith(new EventHandlerTaskThree());
         */

        /**
         * 并行执行
         * 方式一
        disruptor.handleEventsWith(new EventHandlerTaskOne(),new EventHandlerTaskTwo(),new EventHandlerTaskThree());
         */

        /**
         * 并行执行，它们属于不同的EventHandlerGroup
        disruptor.handleEventsWith(new EventHandlerTaskOne());
        disruptor.handleEventsWith(new EventHandlerTaskTwo());
        disruptor.handleEventsWith(new EventHandlerTaskThree());
         */

        /**
         * handler的菱形操作
         *
         *     handler 1
         *                 \handler3
         *                /
         *     handler 2
         * 返回一个EventHandlerGroup对象，里面的两个handler并行执行

        EventHandlerGroup<Trade> tradeEventHandlerGroup = disruptor.handleEventsWith(new EventHandlerTaskOne(), new EventHandlerTaskTwo());

        //在执行handler3
        tradeEventHandlerGroup.then(new EventHandlerTaskThree());
         */

        /**
         * 六边形操作
         *            /------handler2 --------handler3 \
         * handler1 /                                       handler6
                    \                                    /
                     \-------handler4 ----------handler5

         */
        EventHandlerTaskOne handlerTaskOne = new EventHandlerTaskOne();
        EventHandlerTaskTwo handlerTaskTwo  = new EventHandlerTaskTwo();
        EventHandlerTaskThree handlerTaskThree  = new EventHandlerTaskThree();
        EventHandlerTaskFour handlerTaskFour  = new EventHandlerTaskFour();
        EventHandlerTaskFive handlerTaskFive  = new EventHandlerTaskFive();
        EventHandlerTaskSix handlerTaskSix  = new EventHandlerTaskSix();
        disruptor.handleEventsWith(handlerTaskOne).handleEventsWith(handlerTaskTwo,handlerTaskFour);
        disruptor.after(handlerTaskTwo).handleEventsWith(handlerTaskThree);
        disruptor.after(handlerTaskFour).handleEventsWith(handlerTaskFive);
        disruptor.after(handlerTaskThree,handlerTaskFive).handleEventsWith(handlerTaskSix);


        //启动disruptor
        RingBuffer<Trade> ringBuffer = disruptor.start();
        CountDownLatch latch = new CountDownLatch(1);
        long startTime = System.currentTimeMillis();
        executorService.submit(new TradePublisher(latch,disruptor));
        System.out.println("执行耗时: "+(System.currentTimeMillis()-startTime));
        //解放主线程
        latch.await();
        //关闭资源
        disruptor.shutdown();
        executorService.shutdown();
        System.err.println("总耗时: " + (System.currentTimeMillis() - startTime));
    }
}
